/*
 * monitorstream.cc
 *
 *  Created on: Mar 15, 2012
 *      Author: yexijiang
 */

#include "monitorstream.h"

namespace event
{
using namespace std;

ostream& operator<<(ostream &out, ObserveNode &node)
{
	out << "[" << TimeToString(node.timestamp) << ", " << node.value << "]";
	return out;
}

MonitorStream::MonitorStream(const string &name, MonitorStreamType type,
							 int monitor_rate, int range_in_second) :
	name_(name), type_(type), monitor_rate_(monitor_rate), range_in_second_(range_in_second)
{
	pthread_rwlock_init(&rw_mutex_, NULL);
}

string MonitorStream::GetStreamName() const
{
	return name_;
}

string MonitorStream::GetStreamTypeName() const
{
	switch (type_)
	{
	case 0:
		return "CPU";
	case 1:
		return "Main Memory";
	case 2:
		return "Disk Space";
	default:
		return "Error";
	}
}

float MonitorStream::GetStreamRate() const
{
	return monitor_rate_;
}

size_t MonitorStream::Size()
{
	pthread_rwlock_rdlock(&rw_mutex_);
	size_t size =  observe_sequence_.size();
	pthread_rwlock_unlock(&rw_mutex_);
	return size;
}

/*
 * Append a new CPU observation value into sequence.
 */
void MonitorStream::Monitor()
{
	struct timespec sleep_interval;
	sleep_interval.tv_sec = monitor_rate_;
	sleep_interval.tv_nsec = 0;
	struct timespec remain_sleep_interval;
	while (true)
	{
		_ObtainNewObservation();
		nanosleep(&sleep_interval, &remain_sleep_interval);
	}
}

map<string, pair<pthread_t, MonitorStream*> > *StreamManager::streams_ = new map<string, pair<pthread_t, MonitorStream*> >;

bool StreamManager::AttachStream(MonitorStream *stream)
{
	std::string stream_name = stream->GetStreamName();
	if (streams_->find(stream_name) == streams_->end())
	{
		//	suspended streams has negative pthread_t id
		(*streams_)[stream_name] = pair<pthread_t, MonitorStream*>(-1, stream);
		return true;
	}
	else
	{
		return false;
	}
}

const MonitorStream *StreamManager::GetStream(const string &stream_name)
{
	map<string, pair<pthread_t, MonitorStream*> >::const_iterator itr = streams_->find(stream_name);
	if(itr != streams_->end())
		return itr->second.second;
	return NULL;
}

void StreamManager::RunStreamsInThread()
{

	map<string, pair<pthread_t, MonitorStream*> >::iterator itr = streams_->begin();
	for (; itr != streams_->end(); ++itr)
		if (itr->second.first == -1)
			pthread_create(&itr->second.first, NULL, _RunStreamInThread, (void*)itr->second.second);

	fprintf(stderr, "There are %d stream(s) running.\n", streams_->size());

//	for(map<string, pair<pthread_t, MonitorStream*> >::iterator itr = streams_->begin(); itr != streams_->end(); ++itr)
//		if (itr->second.first != -1)
//			pthread_join(itr->second.first, NULL);
}

void *StreamManager::_RunStreamInThread(void *stream)
{
	((MonitorStream*)stream)->Monitor();
}

string CPUMonitorStream::stat_file = "/proc/stat";

CPUMonitorStream::CPUMonitorStream(const string &name, int cpu_index,
								   int monitor_rate, int range_in_second)
	: MonitorStream(name, kCPU, monitor_rate, range_in_second), cpu_index_(cpu_index)
	{}

/*
 * Parse the /proc/stat file to extract the CPU information.
 */
void CPUMonitorStream::_ObtainNewObservation()
{
	int cpu_count = GetCPUCount();
	int line_idx;
	if (cpu_count < cpu_index_)
		return;
	else
		line_idx = cpu_index_ + 1;	//	the stat of CPU is in line <cpu_index_ + 1>

	FILE *fp = fopen(stat_file.c_str(), "r");

	char cpu_id[4];
	unsigned long user_time, nice_time, sys_time, idle_time, iowait_time, int_time, softint_time;
	unsigned long unknown1_time, unknown2_time, unknown3_time;

	for (int i = 0; i <= line_idx; ++i)
		fscanf(fp, "%s %ld %ld %ld %ld %ld %ld %ld %ld %ld %ld\n", cpu_id,
			   &user_time, &nice_time, &sys_time, &idle_time, &iowait_time, &int_time, &softint_time,
			   &unknown1_time, &unknown2_time, &unknown3_time);

	int used_time_before = user_time + nice_time + sys_time;
	int total_time_before = used_time_before + idle_time;

	struct timespec sleep_interval;
	sleep_interval.tv_sec = 0;
	sleep_interval.tv_nsec = 100000000;
	struct timespec remain_sleep_interval;
	nanosleep(&sleep_interval, &remain_sleep_interval);

	rewind(fp);

	for (int i = 0; i <= line_idx; ++i)
		fscanf(fp, "%s %ld %ld %ld %ld %ld %ld %ld %ld %ld %ld\n", cpu_id,
			   &user_time, &nice_time, &sys_time, &idle_time, &iowait_time, &int_time, &softint_time,
			   &unknown1_time, &unknown2_time, &unknown3_time);

	int used_time_after = user_time + nice_time + sys_time;
	int total_time_after = used_time_before + idle_time;

	int used_period = used_time_after - used_time_before;
	int total_period = total_time_after - total_time_before;
	if (total_period < used_period)
		total_period = used_period;

	float ratio = static_cast<float>(used_period) / (total_period);
//	fprintf(stderr, "%s Ratio: %d / %d : %f\n", cpu_id, used_period, total_period, ratio);
	fclose(fp);

	pthread_rwlock_wrlock(&rw_mutex_);
	time_t cur_time;
	time(&cur_time);
	ObserveNode node;
	node.timestamp = cur_time;
	node.value = ratio;
	observe_sequence_.push_back(node);

	//	pop out-of-date observe node
	time_t head_time = observe_sequence_.front().timestamp;
	double time_range = difftime(cur_time, head_time);
	while(time_range > range_in_second_)
	{
		observe_sequence_.pop_front();
		head_time = observe_sequence_.front().timestamp;
		time_range = difftime(cur_time, head_time);
	}
	pthread_rwlock_unlock(&rw_mutex_);
}

std::list<ObserveNode> MonitorStream::GetCurSequence() const
{
	return observe_sequence_;
}

};

